package com.ruyuan.engine.main;
import com.alibaba.fastjson.JSON;
import com.ruyuan.engine.pojo.ClientLog;
import com.ruyuan.engine.pojo.advance.*;
import com.ruyuan.engine.utils.Constants;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.kie.api.io.ResourceType;
import org.kie.api.runtime.KieSession;
import org.kie.internal.utils.KieHelper;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * 动态分区设计
 * */
@Slf4j
public class DynamicPartitionByRuleCondition extends BroadcastProcessFunction<ClientLog, RuleBroadcastTableEnhance, MultipleConditionPartitionRule> {
    public static MapStateDescriptor<String, List<RuleComplexBeanEnhance> > ruleStateDescription = new MapStateDescriptor<>("rules_broadcast_state",TypeInformation.of(String.class),TypeInformation.of(new TypeHint<List<RuleComplexBeanEnhance>>() {}));
    BroadcastState<String, List<RuleComplexBeanEnhance>> broadcastState;

    @Override
    public void processElement(ClientLog clientLog,ReadOnlyContext context,Collector<MultipleConditionPartitionRule> out) throws Exception {
        ReadOnlyBroadcastState<String, List<RuleComplexBeanEnhance>> broadcastState = context.getBroadcastState(ruleStateDescription);
        if (broadcastState == null) return;
        for(Map.Entry<String,List<RuleComplexBeanEnhance>> entry: broadcastState.immutableEntries()) {
            String partitionKey = entry.getKey();
            //ip,province
            String[] partitionKeys = partitionKey.split(",");
            StringBuilder stringBuilder = new StringBuilder();
            /**
             * 通过反射获取到分区字段对应的属性
             * */
            for (String key: partitionKeys) {
                Class<?> clientLogClass = Class.forName("com.ruyuan.engine.pojo.ClientLog");
                Field declaredField = clientLogClass.getDeclaredField(key);
                declaredField.setAccessible(true);
                String fieldValue = (String)declaredField.get(clientLog);
                stringBuilder.append(fieldValue).append(",");
            }
            String partitionValues = stringBuilder.toString().substring(0, stringBuilder.length() - 1);
            clientLog.setPartitionKey(partitionValues);
            /**
             * 生成动态分区的实例
             * */
            MultipleConditionPartitionRule multipleConditionPartitionRule = new MultipleConditionPartitionRule(clientLog,partitionKey,partitionValues);
            out.collect(multipleConditionPartitionRule);
        }
    }

    /**
     * 处理广播的规则数据
     * */
    @Override
    public void processBroadcastElement(RuleBroadcastTableEnhance ruleBroadcastTableEnhance, Context ctx, Collector<MultipleConditionPartitionRule> out) throws Exception {
        broadcastState = ctx.getBroadcastState(ruleStateDescription);
        String operatorRuleType = ruleBroadcastTableEnhance.getOperatorRuleType();
        List<RuleBroadcastTable> rules = ruleBroadcastTableEnhance.getRules();

        /**
         * 遍历所有规则
         * */
        for(RuleBroadcastTable ruleBroadcastTable: rules) {
            //规则字符串
            String ruleConditions = ruleBroadcastTable.getRuleConditions();
            String ruleDrl = ruleBroadcastTable.getRuleDrl();
            Integer ruleId = ruleBroadcastTable.getRuleId();

            //解析规则字符串为Pojo
            RuleComplexBean ruleComplexBean = JSON.parseObject(ruleConditions, RuleComplexBean.class);
            //获取分区条件(IP,UserID)
            String partitionFields = ruleComplexBean.getPartitionFields();
            //状态查询
            List<RuleComplexBeanEnhance> ruleComplexBeanEnhances = broadcastState.get(partitionFields);
            /**
             * 判断：
             *      规则上架
             *      规则下架
             * */
            if (operatorRuleType.equals(Constants.RULE_ACTIVE)) {
                if (ruleComplexBeanEnhances == null) {
                    ruleComplexBeanEnhances = new ArrayList<>();
                }
                KieSession kieSession = new KieHelper().addContent(ruleDrl, ResourceType.DRL).build().newKieSession();
                RuleComplexBeanEnhance ruleComplexBeanEnhance = new RuleComplexBeanEnhance(ruleComplexBean,kieSession);
                ruleComplexBeanEnhances.add(ruleComplexBeanEnhance);
                broadcastState.put(partitionFields,ruleComplexBeanEnhances);

            } else if (operatorRuleType.equals(Constants.RULE_DE_ACTIVE)) {
                //从状态中删除对应的规则
                Iterator<RuleComplexBeanEnhance> ruleComplexBeanEnhanceIterator = ruleComplexBeanEnhances.iterator();
                while (ruleComplexBeanEnhanceIterator.hasNext()) {
                    RuleComplexBeanEnhance ruleComplexBeanEnhance = ruleComplexBeanEnhanceIterator.next();
                    if (ruleComplexBeanEnhance.getRuleComplexBean().getRuleId().equals(ruleId)) ruleComplexBeanEnhanceIterator.remove();
                }
                //删除该分区
                if (ruleComplexBeanEnhances.size() == 0) {
                    broadcastState.remove(partitionFields);
                }
            }else {
                log.error("规则操作异常");
            }
        }
    }
}
